var config = require('./config/index.js');

var kafka = require('kafka-node');
var Producer = kafka.Producer;
var Consumer = kafka.Consumer;
var KeyedMessage = kafka.KeyedMessage;


var client = new kafka.Client(config.zookeeper.uri);
var producer = new Producer(client, {
    requireAcks: 1,
    ackTimeoutMs: 1000
});

var __topic = 't1';

producer.on('error', function(err) {
    console.log('error', err);
});

producer.on('ready', function() {
    console.log('ready');

    // 
    producer.createTopics([__topic], false, function(err, data) {
        if (err) throw err;
        console.log(data);

        // ---create consumer
        var offset = 0;
        var consumer = new Consumer(client, [{
            topic: __topic,
            partition: 0
        }], {
            autoCommit: false,
            groupId: 'kafka-node-group',
            fetchMaxWaitMs: 1000,
            encoding: 'utf8'
        });
        consumer.on('messages', function(msg) {
            console.log('consumer get msg:', msg);
        });

        // function loopFetch() {
        //     console.log('loopFetch');
        //     consumer.fetch({
        //         topic: __topic,
        //         partition: 0,
        //         maxNum: 1
        //     }, function(err, data) {
        //         console.log('loopFetch done');
        //         if (err) return console.log('consumer fetch error:', err);
        //         console.log('consumer fetch ok:', data);
        //         offset++;
        //         setTimeout(loopFetch, 1000);
        //     });
        // }
        // loopFetch();


        // ---producer send msg
        var i = 0;

        function loopSend() {
            i++;
            producer.send([{
                topic: __topic,
                messages: 'hi ' + i,
                partition: 0
            }], function(err, data) {
                if (err) return console.log('send msg err', err);
                console.log('send msg ok', data);
                setTimeout(loopSend, 1000);
            });
        }

        loopSend();
    });
});

// bin/kafka-console-consumer.sh --zookeeper localhost:2181/kafka --from-beginning --topic t1